package com.navinfo.platform.score.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.navinfo.platform.score.pojo.ScoreWeight;
import com.navinfo.platform.score.service.LoadScoreWeightService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;

/**
 * 加载载重权重JSON
 * @author CAI
 */
public class LoadScoreWeightJSONServiceImpl implements LoadScoreWeightService, Serializable {
    static final Logger logger = LoggerFactory.getLogger(LoadScoreWeightJSONServiceImpl.class);

    private String path;

    public String getPath() {
        return path;
    }

    public void setPath(String path) {
        this.path = path;
    }

    @Override
    public JavaPairRDD<String, ScoreWeight> getScoreWeightRdd(SparkSession session) {
        File file = new File(getPath());
        if(!file.exists()) {
            logger.error(" Json文件不存在, PATH:{}",getPath());
            return null;
        }
        Dataset<String> input = session.readStream().json(getPath()).toJSON();
        return input.toJavaRDD().mapPartitionsToPair(
                (PairFlatMapFunction<Iterator<String>, String, ScoreWeight>) LoadScoreWeightJSONServiceImpl::getScoreWeightPair);
    }

    private static Iterator<Tuple2<String, ScoreWeight>> getScoreWeightPair(Iterator<String> iterator) {
        List<Tuple2<String, ScoreWeight>> vehicleInfos = new ArrayList<>();
        ObjectMapper mapper = new ObjectMapper();
        while (iterator.hasNext()) {
            String line = iterator.next();
            ScoreWeight scoreWeight = null;
            try {
                scoreWeight = mapper.readValue(line, ScoreWeight.class);
                String key = scoreWeight.getLoad_code() + "_" + scoreWeight.getLevel2() + "_" + scoreWeight.getModelType();

                vehicleInfos.add(new Tuple2<>(key, scoreWeight));
            } catch (IOException e) {
                e.printStackTrace();
                logger.error(" 解析出错 ");
            }
        }
        return vehicleInfos.iterator();
    }
    @Override
    public Map<String, ScoreWeight> getScoreWeight() {
//        Properties properties = new Properties();
//        properties.put("driver", this.driver);
//        properties.put("user", this.user);
//        properties.put("password", this.pass);
        return new HashMap<>();

    }
}
